using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Linq.Expressions;
using Apache.Cassandra;
using Cassandraemon.Builder;


namespace Cassandraemon
{
    public static class MutateTracker
    {
        #region Insertions (Entities)

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            object entity)
        {
            var col = entity as CassandraEntity<Column>;
            if (col != null)
            {
                AddInsertion(map, col);
                return;
            }
            var rgcol = entity as CassandraEntity<List<Column>>;
            if (rgcol != null)
            {
                AddInsertion(map, rgcol);
                return;
            }
            var scl = entity as CassandraEntity<SuperColumn>;
            if (scl != null)
            {
                AddInsertion(map, scl);
                return;
            }
            var rgscl = entity as CassandraEntity<List<SuperColumn>>;
            if (rgscl != null)
            {
                AddInsertion(map, rgscl);
                return;
            }
            var ccol = entity as CassandraEntity<CounterColumn>;
            if (ccol != null)
            {
                AddInsertion(map, ccol);
                return;
            }
            var rgccol = entity as CassandraEntity<List<CounterColumn>>;
            if (rgccol != null)
            {
                AddInsertion(map, rgccol);
                return;
            }
            var cscl = entity as CassandraEntity<CounterSuperColumn>;
            if (cscl != null)
            {
                AddInsertion(map, cscl);
                return;
            }
            var rgcscl = entity as CassandraEntity<List<CounterSuperColumn>>;
            if (rgcscl != null)
            {
                AddInsertion(map, rgcscl);
                return;
            }

            throw new ArgumentException(string.Format(
                "Entity {0} <{1}> not supported",
                entity,
                entity != null ? entity.GetType().FullName : "null"));
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            CassandraEntity<Column> entity)
        {
            var cs = new ColumnOrSuperColumn { Column = entity.Data };
            AddInsertion(map, entity.ColumnFamily, entity.Key, cs);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            CassandraEntity<SuperColumn> entity)
        {
            var cs = new ColumnOrSuperColumn { Super_column = entity.Data };
            AddInsertion(map, entity.ColumnFamily, entity.Key, cs);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            CassandraEntity<CounterColumn> entity)
        {
            var cs = new ColumnOrSuperColumn { Counter_column = entity.Data };
            AddInsertion(map, entity.ColumnFamily, entity.Key, cs);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            CassandraEntity<CounterSuperColumn> entity)
        {
            var cs = new ColumnOrSuperColumn { Counter_super_column = entity.Data };
            AddInsertion(map, entity.ColumnFamily, entity.Key, cs);
        }

        public static void AddInsertion(
            Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            CassandraEntity<List<Column>> entity)
        {
            var csList = entity.Data.ConvertAll(x => new ColumnOrSuperColumn { Column = x });
            AddInsertion(map, entity.ColumnFamily, entity.Key, csList);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            CassandraEntity<List<SuperColumn>> entity)
        {
            var csList = entity.Data.ConvertAll(x => new ColumnOrSuperColumn { Super_column = x });
            AddInsertion(map, entity.ColumnFamily, entity.Key, csList);
        }

        public static void AddInsertion(
            Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            CassandraEntity<List<CounterColumn>> entity)
        {
            var csList = entity.Data.ConvertAll(x => new ColumnOrSuperColumn { Counter_column = x });
            AddInsertion(map, entity.ColumnFamily, entity.Key, csList);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            CassandraEntity<List<CounterSuperColumn>> entity)
        {
            var csList = entity.Data.ConvertAll(x => new ColumnOrSuperColumn { Counter_super_column = x });
            AddInsertion(map, entity.ColumnFamily, entity.Key, csList);
        }

        #endregion

        #region Insertions (Columns)

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            Column column)
        {
            var mt = new Mutation
            {
                Column_or_supercolumn = new ColumnOrSuperColumn { Column = column }
            };
            Add(map, columnFamily, key, mt);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            SuperColumn column)
        {
            var mt = new Mutation
            {
                Column_or_supercolumn = new ColumnOrSuperColumn { Super_column = column }
            };
            Add(map, columnFamily, key, mt);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            CounterColumn counterColumn)
        {
            var mt = new Mutation
            {
                Column_or_supercolumn = new ColumnOrSuperColumn { Counter_column = counterColumn }
            };
            Add(map, columnFamily, key, mt);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            CounterSuperColumn counterSuperColumn)
        {
            var mt = new Mutation
            {
                Column_or_supercolumn = new ColumnOrSuperColumn { Counter_super_column = counterSuperColumn }
            };
            Add(map, columnFamily, key, mt);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            ColumnOrSuperColumn cs)
        {
            var mt = new Mutation { Column_or_supercolumn = cs };
            Add(map, columnFamily, key, mt);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            IEnumerable<Column> csList)
        {
            var mtList = csList
                .Select(x => new Mutation
                {
                    Column_or_supercolumn = new ColumnOrSuperColumn { Column = x }
                });
            AddRange(map, columnFamily, key, mtList);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            IEnumerable<SuperColumn> csList)
        {
            var mtList = csList
                .Select(x => new Mutation
                {
                    Column_or_supercolumn = new ColumnOrSuperColumn { Super_column = x }
                });
            AddRange(map, columnFamily, key, mtList);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            IEnumerable<CounterColumn> csList)
        {
            var mtList = csList
                .Select(x => new Mutation
                {
                    Column_or_supercolumn = new ColumnOrSuperColumn { Counter_column = x }
                });
            AddRange(map, columnFamily, key, mtList);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            IEnumerable<CounterSuperColumn> csList)
        {
            var mtList = csList
                .Select(x => new Mutation
                {
                    Column_or_supercolumn = new ColumnOrSuperColumn { Counter_super_column = x }
                });
            AddRange(map, columnFamily, key, mtList);
        }

        public static void AddInsertion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            IEnumerable<ColumnOrSuperColumn> csList)
        {
            var mtList = csList
                .Select(x => new Mutation { Column_or_supercolumn = x });
            AddRange(map, columnFamily, key, mtList);
        }

        #endregion

        #region Deletions

        public static void AddDeletion(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            LambdaExpression expression)
        {
            var builder = new CassandraDeletionBuilder();
            builder.Build(expression);
            Add(map, builder.ColumnFamily, builder.Key, builder.Deletion);
        }

        public static void Add(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            CassandraBinary key,
            Deletion deletion)
        {
            var mt = new Mutation { Deletion = deletion };
            Add(map, columnFamily, key, mt);
        }

        #endregion

        #region Mutations

        public static void Add(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            Mutation mutation)
        {
            SetTimeStamp(mutation);
            var rgmut = GetList(map, columnFamily, key);
            rgmut.Add(mutation);
        }

        public static void AddRange(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key,
            IEnumerable<Mutation> mutations)
        {
            SetTimeStamp(mutations);
            var rgmut = GetList(map, columnFamily, key);
            rgmut.AddRange(mutations);
        }

        public static void SetTimeStamp(this Mutation mutation)
        {
            long now = TimeGenerator.GetUnixTime();
            SetTimeStamp(mutation, now);
        }

        public static void SetTimeStamp(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map)
        {
            SetTimeStamp(map.Mutations());
        }

        public static void SetTimeStamp(this IEnumerable<Mutation> mutations)
        {
            long now = TimeGenerator.GetUnixTime();
            foreach (var m in mutations)
            {
                SetTimeStamp(m, now);
            }
        }

        static void SetTimeStamp(
            this Mutation m,
            long timestamp)
        {
            var cob = m.Column_or_supercolumn;
            if (cob != null)
            {
                var col = cob.Column;
                if (col != null)
                {
                    if (col.Timestamp == default(long))
                    {
                        col.Timestamp = timestamp;
                    }
                }

                var spc = cob.Super_column;
                if (spc != null)
                {
                    foreach (var c in spc.Columns)
                    {
                        if (c.Timestamp == default(long))
                        {
                            c.Timestamp = timestamp;
                        }
                    }
                }
            }

            var del = m.Deletion;
            if (del != null)
            {
                if (del.Timestamp == default(long))
                {
                    del.Timestamp = timestamp;
                }
            }
        }

        #endregion

        #region Low-Level Utilities

        public static int CountMutations(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map)
        {
            return map
                .SelectMany(rg => rg.Value)
                .SelectMany(rg => rg.Value)
                .Count();
        }

        public static IEnumerable<Mutation> Mutations(
            this Dictionary<byte[], Dictionary<string, List<Mutation>>> map)
        {
            return map
                .SelectMany(rg => rg.Value)
                .SelectMany(rg => rg.Value);
        }

        static List<Mutation> GetList(
            Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
            string columnFamily,
            byte[] key)
        {
            if (key == null)
            {
                throw new ArgumentNullException("key");
            }
            if (string.IsNullOrEmpty(columnFamily))
            {
                throw new ArgumentNullException("columnFamily");
            }

            Dictionary<string, List<Mutation>> rgmutByColumnFamily;
            if (!map.TryGetValue(key, out rgmutByColumnFamily))
            {
                rgmutByColumnFamily = new Dictionary<string, List<Mutation>>();
                map.Add(key, rgmutByColumnFamily);
            }

            List<Mutation> rgmut;
            if (!rgmutByColumnFamily.TryGetValue(columnFamily, out rgmut))
            {
                rgmut = new List<Mutation>();
                rgmutByColumnFamily.Add(columnFamily, rgmut);
            }

            return rgmut;
        }

        #endregion

        #region Mutations Reductions

		internal static void RemoveIfNoChange(
			this Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
			CassandraContext ctx,
			Func<int, bool> progress)
		{
			RemoveIfNoChange(
				map,
				(columnFamily, key, superColumnName, columnNames) =>
				{
					var rgColumnName = columnNames.ToList();
					if (rgColumnName.Count == 0)
					{
						return null;
					}

					var parent = new ColumnParent()
					{
						Column_family = columnFamily,
						Super_column = superColumnName,
					};
					var predicate = new SlicePredicate()
					{
						Column_names = rgColumnName
					};
					try
					{
						var rgcob = ctx.Operate(() =>
						{
							var client = ctx.Client;
							var sw = Stopwatch.StartNew();
							var result = client.get_slice(
								key,
								parent,
								predicate,
								ctx.ConsistencyLevel);
                            ctx.OnExecuted(new CassandraExecuteEventArgs(
                                result,
                                sw.Elapsed,
                                client,
                                "get_slice",
                                columnFamily));
							return result;
						});

						return rgcob;
					}
					catch (TimedOutException err)
					{
						// Read failure can result writing the same data at worst,
						// which will be removed anyway on compaction,
						// so it's better to ignore the failure and continue.
#if TRACE
						CassandraContext.TraceWarning(
							"RemoveIfNoChange: Read failed, ignored: {0}",
							err);
#endif
						return null;
					}
				}, progress);
		}

		static void RemoveIfNoChange(
			Dictionary<byte[], Dictionary<string, List<Mutation>>> map,
			Func<string, byte[], byte[], IEnumerable<byte[]>,
				IEnumerable<ColumnOrSuperColumn>> reader,
			Func<int, bool> progress)
		{
			int imutProgres = 0;

            var rgkeyToRemove = new List<byte[]>();
            var rgcfToRemove = new List<string>();
			var rgcobByName = new Dictionary<byte[], ColumnOrSuperColumn>(
                SequenceComparer<byte>.Default);
            foreach (var kvpKey in map)
            {
                var key = kvpKey.Key;
                var rgrgmutByCf = kvpKey.Value;
                foreach (var kvpCfRgmut in rgrgmutByCf)
                {
                    var rgmut = kvpCfRgmut.Value;
                    if (rgmut == null)
                    {
                        continue;
                    }

					var columnFamily = kvpCfRgmut.Key;
                    var cmutBefore = rgmut.Count;
                    if (cmutBefore <= 0)
                    {
						// Already no mutations.
						// This isn't supposed to happen, but just in case.
						rgcfToRemove.Add(columnFamily);
                        continue;
                    }

					if (progress != null)
					{
						if (progress(imutProgres)) { return; }
						imutProgres += cmutBefore;
					}

                    // Removes Mutations that will be overwritten
                    // in the following Mutations.

					rgcobByName.Clear();
					if (!RemoveDuplicate(rgmut, rgcobByName))
                    {
                        continue;
                    }

                    // If the value is equal to the value in the database,
                    // the mutation is not necessary.

					if (reader != null)
					{
						RemoveIfNoChange(rgmut,
							columnFamily, key,
							rgcobByName,
							reader);
					}

                    var cmutAfter = rgmut.Count;
                    if (cmutAfter == 0)
                    {
                        rgcfToRemove.Add(columnFamily);
                    }
                }

                foreach (var cf in rgcfToRemove)
                {
                    rgrgmutByCf.Remove(cf);
                }
                rgcfToRemove.Clear();

                if (rgrgmutByCf.Count == 0)
                {
                    rgkeyToRemove.Add(key);
                }
            }

            foreach (var key in rgkeyToRemove)
            {
                map.Remove(key);
            }
        }


		static void RemoveIfNoChange(
			List<Mutation> rgmut,
			string columnFamily,
			byte[] key,
			Dictionary<byte[], ColumnOrSuperColumn> rgcobByName,
			Func<string, byte[], byte[], IEnumerable<byte[]>,
				IEnumerable<ColumnOrSuperColumn>> reader)
		{
			if (reader == null)
			{
				return;
			}

			// Read all Columns in the mutations from the database.
			// For SuperColumns, there are two options: 1) read all sub columns,
			// or 2) read affected sub columns only. While 2 gives less data
			// transfer, 1 gives less round-trips.
			// Current implementation is 1, so if you're modifying a few sub columns
			// from a large set of sub columns, it's inefficient.

			if (rgcobByName.Count <= 0)
			{
				return;
			}
			var rgColumnNameToMutate = rgcobByName
				.Select(kvp => kvp.Key);
			var rgcobFromDB = reader(columnFamily, key, null,
				rgColumnNameToMutate);
			if (rgcobFromDB == null)
			{
				return;
			}

			var columnNameComparer = SequenceComparer<byte>.Default;
			var rgcobByNameFromDB = rgcobFromDB
				.ToDictionary(c => c.NameOrDefault(), columnNameComparer);

			// Run through mutations and see if each of them should be left
			// or are safe to remove.

			for (int imut = rgmut.Count - 1; imut >= 0; imut--)
			{
				var mut = rgmut[imut];
				var cob = mut.Column_or_supercolumn;
				if (cob != null)
				{
					// For Columns, just compare the timestamp/value in DB.

					var col = cob.Column;
					if (col != null)
					{
						ColumnOrSuperColumn cobFromDb;
						if (rgcobByNameFromDB.TryGetValue(col.Name, out cobFromDb) &&
							!IsChanged(col, cobFromDb.Column))
						{
							rgmut.RemoveAt(imut);
						}
					}

					// For SuperColumns, compare sub columns.

					var sco = cob.Super_column;
					if (sco != null)
					{
						ColumnOrSuperColumn cobFromDB;
						if (rgcobByNameFromDB.TryGetValue(sco.Name, out cobFromDB))
						{
							var rgSubColumnsByNameFromDB = cobFromDB.Super_column.Columns
								.ToDictionary(c => c.Name, columnNameComparer);
							var rgcol = sco.Columns;
							for (int icol = rgcol.Count - 1; icol >= 0; icol--)
							{
								col = rgcol[icol];
								Column colFromDB;
								if (rgSubColumnsByNameFromDB.TryGetValue(col.Name, out colFromDB) &&
									!IsChanged(col, colFromDB))
								{
									rgcol.RemoveAt(icol);
								}
							}

							// If no subcolumns are left, remove the mutation.

							if (rgcol.Count == 0)
							{
								rgmut.RemoveAt(imut);
							}
						}
					}
					
					// For CounterColumns, just compare the value in DB.
					
					var ccol = cob.Counter_column;
					if (ccol != null)
					{
						ColumnOrSuperColumn cobFromDb;
						if (rgcobByNameFromDB.TryGetValue(ccol.Name, out cobFromDb) &&
						    !IsChanged(ccol, cobFromDb.Counter_column))
						{
							rgmut.RemoveAt(imut);
						}
					}
					
					// For CounterSuperColumns, compare sub columns.
					
					var csco = cob.Counter_super_column;
					if (csco != null)
					{
						ColumnOrSuperColumn cobFromDB;
						if(rgcobByNameFromDB.TryGetValue(csco.Name, out cobFromDB))
						{
							var rgSubColumnsByNameFromDB = cobFromDB.Counter_super_column.Columns
								.ToDictionary(c => c.Name, columnNameComparer);
							var rgcol = csco.Columns;
							for (int icol = rgcol.Count - 1; icol >= 0; icol--)
							{
								ccol = rgcol[icol];
								CounterColumn colFromDB;
								if (rgSubColumnsByNameFromDB.TryGetValue(ccol.Name, out colFromDB) &&
								    !IsChanged(ccol, colFromDB))
								{
									rgcol.RemoveAt(icol);
								}
							}
							
							// If no subcolumns are left, remove the mutation.
							
							if (rgcol.Count == 0)
							{
								rgmut.RemoveAt(imut);
							}
						}
					}
				}
			}
		}

		static bool IsChanged(Column col, Column colFromDB)
		{
			if (colFromDB == null)
			{
				return true;
			}

			var timestamp = col.Timestamp;
			if (timestamp != 0 && timestamp <= colFromDB.Timestamp)
			{
				return false;
			}

			if (col.Value.SequenceEqual(colFromDB.Value))
			{
				return false;
			}

			return true;
		}

		static bool IsChanged(CounterColumn col, CounterColumn colFromDB)
		{
			if (colFromDB == null)
			{
				return true;
			}

			if (col.Value == colFromDB.Value)
			{
				return false;
			}

			return true;
		}

        static bool RemoveDuplicate(
            List<Mutation> rgmut,
			Dictionary<byte[], ColumnOrSuperColumn> rgcobByName)
        {
            for (int imut = rgmut.Count - 1; imut >= 0; imut--)
            {
                var mut = rgmut[imut];
                var del = mut.Deletion;
                if (del != null)
                {
                    var pred = del.Predicate;
                    if (pred != null)
                    {
						var rgColumnName = pred.Column_names;
						if (rgColumnName != null)
                        {
                            var fAdd = false;
							foreach (var columnName in rgColumnName)
							{
								if (!rgcobByName.ContainsKey(columnName))
								{
									rgcobByName.Add(columnName, null);
                                    fAdd = true;
                                }
                            }
                            if (!fAdd)
                            {
                                rgmut.RemoveAt(imut);
                            }
                            continue;
                        }
                    }

                    // Other formats of Deletion are not supported
                    // e.g. Timestamp, Super_column, Slice_range, etc.

                    return false;
                }

                // If the column will be overwritten in the following mutations,
                // the mutation is not necessary.

                var cob = mut.Column_or_supercolumn;
				if (cob != null)
				{
					var name = cob.NameOrDefault();

					ColumnOrSuperColumn cobNewer;
					if (!rgcobByName.TryGetValue(name, out cobNewer))
					{
						rgcobByName.Add(name, cob);
					}
					else
					{
						var sco = cob.Super_column;
						if (sco != null && cobNewer != null)
						{
							// If multiple mutations exist for the same SuperColumn,
							// merge them before removing earlier one.

							var scoNewer = cobNewer.Super_column;
							var merged = scoNewer.Columns
								.Union(sco.Columns, ColumnExtensions.ColumnComparerByName)
								.ToList();
							if (scoNewer.Columns.Count != merged.Count)
							{
								scoNewer.Columns = merged;
							}
						}
						
						var csco = cob.Counter_super_column;
						if (csco != null && cobNewer != null)
						{
							// If multiple mutations exist for the same CounterSuperColumn,
							// merge them before removing earlier one.
							
							var cscoNewer = cobNewer.Counter_super_column;
							var merged = cscoNewer.Columns
								.Union(csco.Columns, CounterColumnExtensions.CounterColumnComparerByName)
								.ToList();
							if (cscoNewer.Columns.Count != merged.Count)
							{
								cscoNewer.Columns = merged;
							}
						}

						rgmut.RemoveAt(imut);
					}

					continue;
				}

				throw new FormatException("Unknown Mutation format");
            }

            return true;
        }

        #endregion
    }
}